package com.stm.bi.etl;

import com.stm.bi.constant.DimensionConstants;
import com.stm.bi.core.util.DateUtils;
import com.stm.bi.enums.DateType;
import com.stm.bi.repository.AppOrganizationRepository;
import com.stm.bi.repository.DorisLoadRepository;
import com.stm.bi.repository.EtlRecordRepository;
import com.stm.bi.repository.dataobject.EtlRecordDO;
import com.stm.framework.core.util.SpringContextUtil;
import com.stm.framework.dbcp.JdbcService;
import com.stm.ucenter.dto.vo.org.OrganizationVO;
import jakarta.annotation.Resource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @Author chenyingjun
 * @Desc
 * @Date 2024/11/29 10:58
 */
public abstract class AbstractDataEtlTask implements Runnable{
    protected static final Logger log = LoggerFactory.getLogger(AbstractDataEtlTask.class);

    @Resource
    protected JdbcService jdbcService;
    @Resource
    private AppOrganizationRepository appOrganizationRepository;
    @Resource
    private EtlRecordRepository etlRecordRepository;
    @Resource
    protected DorisLoadRepository dorisLoadRepository;

    public abstract boolean isTenantLine();

    abstract public boolean syncWithTimeRange();

    /**
     * 获取任务名称
     *
     * 此方法用于获取当前任务的名称，每个任务都有一个唯一的名称用于标识
     * 子类必须实现此方法以提供具体的任务名称
     *
     * @return 当前任务的名称
     */
    public abstract String getTaskName();

    /**
     * 获取ETL日期类型列表
     *
     * 此方法用于获取当前任务中涉及的所有ETL日期类型这些日期类型
     * 用于指示任务在不同时间维度上如何处理数据，例如按天、按月等
     * 子类必须实现此方法以提供具体的ETL日期类型列表
     *
     * @return 包含所有ETL日期类型的列表
     */
    public abstract List<DateType> getEtlDateTypes();

    public abstract int getDelayDays();

    public void run(){
        execute();
    }

    public void execute() {
        LocalDateTime pointDateTime = LocalDateTime.now();
        int delayDays = getDelayDays();
        if(delayDays > 0){
            pointDateTime = pointDateTime.minusDays(delayDays - 1);
            pointDateTime = pointDateTime.minusHours(pointDateTime.getHour()).minusMinutes(pointDateTime.getMinute()).minusSeconds(pointDateTime.getSecond()+1).minusNanos(pointDateTime.getNano());
        }
        if(isTenantLine()){
            List<OrganizationVO> organizations = appOrganizationRepository.selectOrgList();
            for(OrganizationVO org : organizations){
                if(syncWithTimeRange()){
                    extracted(org,pointDateTime);
                }else{
                    extracted(org);
                }
            }
        }else{
            if(syncWithTimeRange()){
                extracted(new OrganizationVO(),pointDateTime);
            }else{
                extracted(new OrganizationVO());
            }
        }
    }

    private void extracted(OrganizationVO org) {
        AbstractDataEtlTask dataEtlTask = SpringContextUtil.getBean(getClass());
        EtlRecordDO etlRecord = etlRecordRepository.getAndCreateIfAbsent(org.getId(), org.getName(),getTaskName());
        List<DateType> dateTypes = getEtlDateTypes();
        if(dateTypes == null || dateTypes.isEmpty()){
            List<Map> dataList = dataEtlTask.extract(org.getId(), org.getNo(),null, null, null);
            if(dataList.isEmpty() == false){
                setCommonFieldValues(org, dataList);
                dataEtlTask.load(org.getId(), org.getNo(),dataList);
            }
        }else{
            for(DateType dateType : dateTypes){
                List<Map> dataList = dataEtlTask.extract(org.getId(), org.getNo(),dateType, null, null);
                dataList.forEach(rowData -> {
                    rowData.put("org_id", org.getId());
                    rowData.put("org_no", org.getNo());
                    rowData.put("date_type",dateType.getValue());
//                    if(DateType.WEEK.equals(dateType)){
//                        String date = (String)rowData.get("date");
//                        String[] arrays = date.split("-");
//                        int week = Integer.parseInt(arrays[1]);
//                        if(week<10){
//                            rowData.put("date",arrays[0]+"-0"+arrays[1]);
//                        }
//                    }
                });
                if(dataList.isEmpty() == false){
                    dataList = dataEtlTask.transform(org.getId(), org.getNo(),dataList);
                    dataList.forEach(map->{
                        map.put("org_no", org.getNo());
                        map.put("org_id", org.getId());
                    });
                    dataEtlTask.load(org.getId(), org.getNo(),dataList);
                }
            }
        }
        etlRecord.setLastEtlTime(LocalDateTime.now());
        etlRecordRepository.update(etlRecord);
    }


    private void extracted(OrganizationVO org,LocalDateTime pointDateTime) {
        AbstractDataEtlTask dataEtlTask = SpringContextUtil.getBean(getClass());
        EtlRecordDO etlRecord = etlRecordRepository.getAndCreateIfAbsent(org.getId(), org.getName(),getTaskName());
        List<DateType> dateTypes = getEtlDateTypes();
        LocalDateTime lastEtlTime = etlRecord.getLastEtlTime();

        while (lastEtlTime.isBefore(pointDateTime)){
            LocalDateTime endDateTime = lastEtlTime.plusYears(1).minusSeconds(1);
            if(!endDateTime.isBefore(pointDateTime)){
                endDateTime = pointDateTime;
            }
            if(dateTypes == null || dateTypes.isEmpty()){
                LocalDateTime startDateTime = LocalDateTime.of(lastEtlTime.getYear(),1,1,0,0,0);
                List<Map> dataList = dataEtlTask.extract(org.getId(), org.getNo(),null, startDateTime, endDateTime);
                if(dataList.isEmpty() == false){
                    setCommonFieldValues(org, dataList);
                    dataEtlTask.load(org.getId(), org.getNo(),dataList);
                }
            }else{
                for(DateType dateType : dateTypes){
                    LocalDateTime startDateTime = getStartDateTime(lastEtlTime, dateType);

                    List<String> etlDateList = DateUtils.getETLDateList(startDateTime, endDateTime, dateType);
                    for (String etlDate : etlDateList) {
                        LocalDateTime etlDateTime = DateUtils.parseStringToLocalDateTime(etlDate, dateType);
                        List<Map> dataList = dataEtlTask.extract(org.getId(), org.getNo(),dateType, etlDateTime, etlDateTime);
                        setCommonFieldValues(org, dataList);
//                        dataList.forEach(rowData -> {
//                            rowData.put("date_type",dateType.getValue());
//                            if(DateType.WEEK.equals(dateType)){
//                                String date = (String)rowData.get("date");
//                                String[] arrays = date.split("-");
//                                int week = Integer.parseInt(arrays[1]);
//                                if(week<10){
//                                    rowData.put("date",arrays[0]+"-0"+arrays[1]);
//                                }
//                            }
//                        });
                        if(dataList.isEmpty() == false){
                            dataList = dataEtlTask.transform(org.getId(), org.getNo(),dataList);
                            setCommonFieldValues(org, dataList);
                            dataEtlTask.load(org.getId(), org.getNo(),dataList);
                        }
                    }


                }
            }
            etlRecord.setLastEtlTime(endDateTime);
            etlRecordRepository.update(etlRecord);
            lastEtlTime = endDateTime.plusSeconds(1);
        }
    }


    private static LocalDateTime getStartDateTime(LocalDateTime lastEtlTime, DateType dateType) {
        LocalDateTime startDateTime = null;
        switch (dateType){
            case YEAR:
                startDateTime = LocalDateTime.of(lastEtlTime.getYear(),1,1,0,0,0);
                break;
            case MONTH:
                startDateTime = LocalDateTime.of(lastEtlTime.getYear(), lastEtlTime.getMonth(),1,0,0,0);
                break;
            case WEEK:
                int dayOfYear = lastEtlTime.getDayOfYear();
                int dayOfWeek = lastEtlTime.getDayOfWeek().getValue();
                if(dayOfWeek <= dayOfYear){
                    startDateTime = lastEtlTime.minusDays(dayOfWeek - 1);
                }else{
                    startDateTime = lastEtlTime.minusDays(dayOfYear);
                }
                startDateTime = LocalDateTime.of(startDateTime.getYear(),startDateTime.getMonth(),startDateTime.getDayOfMonth(),0,0,0);
                break;
            case DAY:
                startDateTime = LocalDateTime.of(lastEtlTime.getYear(), lastEtlTime.getMonth(), lastEtlTime.getDayOfMonth(),0,0,0);
                break;
        }
        return startDateTime;
    }

    private static void setCommonFieldValues(OrganizationVO org, List<Map> dataList) {
        dataList.forEach(data -> {
            data.put("org_id", org.getId());
            data.put("org_no", org.getNo());
            data.put("org_name", org.getName());
            data.put("province_code", data.get("province_id"));
            data.put("city_code", data.get("city_id"));
            data.put("district_code", data.get("city_id"));
        });
    }

    /**
     * 根据指定日期类型和时间范围提取数据
     * 此方法用于从某个数据源（具体实现取决于子类）提取在指定日期类型和时间范围内的数据
     * 例如，可以根据"日"类型提取从某个开始时间到结束时间内的所有记录
     *
     * @param dateType 日期类型，例如日、月、年等，用于指定数据的时间粒度
     * @param startDateTime 提取数据的开始时间
     * @param endDateTime 提取数据的结束时间
     * @return 返回一个包含多条数据的列表，每条数据是一个Map对象
     */
    public abstract List<Map> extract(Long orgId,String orgNo,DateType dateType, LocalDateTime startDateTime, LocalDateTime endDateTime);

    /**
     * 将提取到的数据列表进行转换处理
     * 该方法接收一个数据列表，并对其进行转换处理，转换的规则和逻辑由子类具体实现
     * 通常用于将数据从一种格式转换为另一种格式，以便于后续的加载操作
     *
     * @param dataList 待转换的数据列表，每个数据项是一个Map对象
     * @return 返回转换后的数据列表，每个数据项仍然是一个Map对象
     */
    public abstract List<Map> transform(Long orgId,String orgNo,List<Map> dataList);

    /**
     * 将转换后的数据加载到目标数据源或数据结构中
     * 此方法负责将转换后的数据列表加载到指定的目标位置，如数据库、文件或其他数据结构
     * 具体的加载逻辑由子类实现，确保数据能够被正确地插入或更新
     *
     * @param dataList 待加载的数据列表，每个数据项是一个Map对象
     */
    public abstract void load(Long orgId,String orgNo,List<Map> dataList);


    protected static String formatDateTime(String columnName, DateType dateType) {
        String dateField = null;
        if(dateType.equals(DateType.DAY)){
            dateField = "date_format("+columnName+",'%Y-%m-%d')";
        }else if(dateType.equals(DateType.WEEK)){
            dateField = "concat(date_format("+columnName+",'%Y'),'-',week("+columnName+",1)+1)";
        }else if(dateType.equals(DateType.MONTH)){
            dateField = "date_format("+columnName+",'%Y-%m')";
        }else if(dateType.equals(DateType.YEAR)){
            dateField = "date_format("+columnName+",'%Y')";
        }
        return dateField;
    }
}
